

package com.hazelcast.internal.util.phonehome;

import com.hazelcast.config.Config;
import com.hazelcast.config.NativeMemoryConfig;
import com.hazelcast.instance.impl.Node;
import com.hazelcast.internal.memory.MemoryStats;
import com.hazelcast.map.LocalMapStats;
import com.hazelcast.map.impl.MapService;
import com.hazelcast.replicatedmap.impl.ReplicatedMapService;
import com.hazelcast.spi.impl.NodeEngine;

import java.util.stream.Stream;

import static com.hazelcast.internal.util.phonehome.PhoneHomeMetrics.*;

class StorageMetricsProvider implements MetricsProvider {

    @Override
    public void provideMetrics(Node node, MetricsCollectionContext context) {
        memory(node, context);
        tieredStorage(node.getConfig(), context);
        dataMemoryCost(node.getNodeEngine(), context);
    }

    private void memory(Node node, MetricsCollectionContext context) {
        NativeMemoryConfig nativeMemoryConfig = node.getConfig().getNativeMemoryConfig();

        boolean isHdEnabled = nativeMemoryConfig.isEnabled();
        context.collect(HD_MEMORY_ENABLED, isHdEnabled);

        MemoryStats memoryStats = node.getNodeExtension().getMemoryStats();
        if (isHdEnabled) {
            context.collect(MEMORY_USED_NATIVE_SIZE, memoryStats.getUsedNative());
        }
        context.collect(MEMORY_USED_HEAP_SIZE, memoryStats.getUsedHeap());
        context.collect(MEMORY_FREE_HEAP_SIZE, memoryStats.getFreeHeap());
    }

    private void tieredStorage(Config config, MetricsCollectionContext context) {
        boolean tieredStorageEnabled = config.getMapConfigs().values().stream().anyMatch(mapConfig -> mapConfig.getTieredStoreConfig().isEnabled());
        context.collect(TIERED_STORAGE_ENABLED, tieredStorageEnabled);
    }

    private void dataMemoryCost(NodeEngine nodeEngine, MetricsCollectionContext context) {
        MapService mapService = nodeEngine.getService(MapService.SERVICE_NAME);
        ReplicatedMapService replicatedMapService = nodeEngine.getService(ReplicatedMapService.SERVICE_NAME);
        long totalEntryMemoryCost = Stream.concat(mapService.getStats().values().stream(), replicatedMapService.getStats().values().stream()).mapToLong(LocalMapStats::getOwnedEntryMemoryCost).sum();
        context.collect(DATA_MEMORY_COST, totalEntryMemoryCost);
    }
}
